package com.gandong8.trace.tool.config.aliyun.ons;

import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.bean.ConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import com.gandong8.trace.tool.config.GlobalProperties;
import com.gandong8.trace.tool.config.aliyun.ons.properties.OnsConsumerCommonData;
import com.gandong8.trace.tool.config.aliyun.ons.properties.OnsConsumerProperties;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
@Configuration
@EnableConfigurationProperties({OnsConsumerProperties.class , GlobalProperties.class})
@Slf4j
public class OnsConsumerConf {
    @Autowired
    OnsConsumerProperties publicProp;
    @Autowired
    GlobalProperties globalProperties;

    private void dealConf(OnsConsumerCommonData myProp) {
        if(myProp.getGroupId() == null) {
            myProp.setGroupId(publicProp.getGroupId());
        }
        if(myProp.getNamesrvAddr() == null) {
            myProp.setNamesrvAddr(publicProp.getNamesrvAddr());
        }
        if(myProp.getAccessKey() == null) {
            myProp.setAccessKey(publicProp.getAccessKey());
        }
        if(myProp.getAccessSecret() == null) {
            myProp.setAccessSecret(publicProp.getAccessSecret());
        }
    }


    public ConsumerBean createConsumer(OnsConsumerCommonData myConf, MessageListener listener) {
        if(!globalProperties.getOnsConsumerOpen()) {
            System.err.println("ons consumer not allowed");
            return null;
        }
        dealConf(myConf);
        Properties properties = initProperties(myConf);
        ConsumerBean consumerBean = new ConsumerBean();
        consumerBean.setProperties(properties);
        Subscription subscription = new Subscription();
        subscription.setTopic(myConf.getTopic());
        Map<Subscription, MessageListener> map = new HashMap<>();
        map.put(subscription, listener);
        consumerBean.setSubscriptionTable(map);
        consumerBean.start();
        log.info("----ons consumer start with topic:{}-----", myConf.getTopic());
        return consumerBean;
    }

    private Properties initProperties(OnsConsumerCommonData myConf) {
        Properties prop = new Properties();
        prop.put(PropertyKeyConst.AccessKey, myConf.getAccessKey());
        prop.put(PropertyKeyConst.SecretKey, myConf.getAccessSecret());
        prop.setProperty(PropertyKeyConst.SendMsgTimeoutMillis, myConf.getTimeOut());
        prop.put(PropertyKeyConst.NAMESRV_ADDR, myConf.getNamesrvAddr());
        prop.put(PropertyKeyConst.GROUP_ID, myConf.getGroupId());
        return prop;
    }
}
